package com.spica.platform.kafka.producer.autoconfigure;

import com.spica.platform.common.utils.JsonUtil;
import com.spica.platform.common.utils.StringUtil;
import com.spica.platform.kafka.producer.properties.KafkaProducerProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
@EnableConfigurationProperties(KafkaProducerProperties.class)
@Slf4j
public class KafkaProducerAutoConfiguration {

    @Bean
    public Producer<String, String> producer(KafkaProducerProperties producerProperties) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producerProperties.getBootstrapServers());
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, String.valueOf(producerProperties.isEnableIdempotence()));
        if (StringUtil.isNotBlank(producerProperties.getTransactionalId())) {
            properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, producerProperties.getTransactionalId());
        }
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, producerProperties.getKeySerializer());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, producerProperties.getValueSerializer());
        if(!producerProperties.isEnableIdempotence()){
            properties.put(ProducerConfig.ACKS_CONFIG, producerProperties.getAcks());
        }
        log.info("initing stringStringKafkaProducer using properties : {}", JsonUtil.toJsonStr(properties));
        KafkaProducer<String, String> stringStringKafkaProducer = new KafkaProducer<>(properties);
        log.info("init stringStringKafkaProducer successfully {} using properties : {}", stringStringKafkaProducer, JsonUtil.toJsonStr(properties));
        return stringStringKafkaProducer;
    }
}
